package com.aotain.metis.bolt;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;

/**
 * 
 * @ClassName: UDIIBotsBolt 
 * @Description: TODO(这里用一句话描述这个类的作用)僵尸主机 
 * @author 程彬
 * @date 2016年11月8日 
 *
 */
public class BotsBolt extends AbstractBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * 固定源60s内握手成功次数  目标|源|端口
	 */
	private HashMap<String,Integer> fixedSourceMap = new HashMap<String,Integer>();

	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> upStreamoctets = new HashMap<String,Long>();

	/**
	 * 上行包数
	 */
	private HashMap<String,Long> upStreampacket = new HashMap<String,Long>();

	/**
	 * 受攻击的记录map
	 */
	private HashMap<String,String> abnormalStreamMap = new HashMap<String,String>();

	/**
	 * 攻击开始时间
	 */
	private HashMap<String,Long> attackStartMap = new HashMap<String,Long>();

	/**
	 * 攻击持续的上次更新时间
	 */
	private HashMap<String,Long> attackLasttimeMap = new HashMap<String,Long>();

	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;


	@Override
	public void execute(Tuple tuple) {
		
		try {
			if(isTickTuple(tuple)) {
				Logger.getLogger(BotsBolt.class).info(String.format("###DEBUG:UDIIBots START EMIT [%s]",tuple.getSourceStreamId()));
				emitCountingData(collector);
				Logger.getLogger(BotsBolt.class).info(String.format("###DEBUG:UDIIBots END EMIT [%s]",tuple.getSourceStreamId()));
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("UDIIBotsBolt throws an exception!",e);
		}
	}

//	private boolean isTickTuple(Tuple tuple) {
//		return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(
//				Constants.SYSTEM_TICK_STREAM_ID);   
//	}

	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{

			String dip = tuple.getStringByField("dip");
			String sip = tuple.getStringByField("sip");
			String idc = tuple.getStringByField("idcno");
			String type = tuple.getStringByField("type");
			Integer status = tuple.getIntegerByField("status");
			String gis =  new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
			String dport = tuple.getStringByField("dport");
			Long upStreamOctets = tuple.getLongByField("upstreamoctets");
			Long upStreamPacket = tuple.getLongByField("upstreampacket");


			int fixedSourceCount = 1;
			String fixedMapKey = dip + "|" + sip + "|" + dport + "|" + gis + "|" + idc;
//			//如果是tcp (SYN ACK FIN)则计数
//			if("6".equals(type)) {

				if(fixedSourceMap.containsKey(fixedMapKey)) {
					fixedSourceCount = fixedSourceMap.get(fixedMapKey) + 1;
				}

				if(upStreamoctets.containsKey(fixedMapKey)) {
					upStreamOctets += upStreamoctets.get(fixedMapKey) ;
				}

				if(upStreampacket.containsKey(fixedMapKey)) {
					upStreamPacket += upStreampacket.get(fixedMapKey);
				}
				fixedSourceMap.put(fixedMapKey, fixedSourceCount);
				upStreamoctets.put(fixedMapKey, upStreamOctets);
				upStreampacket.put(fixedMapKey, upStreamPacket);
//			}

		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("BotsBolt=====countInLocal=========ERROR=================",ex);
		}
	}

	private void emitCountingData(OutputCollector collector) {

		try {
			//		InsertIntoData();
			for(Entry<String, Integer> entry : fixedSourceMap.entrySet()) {

				String entryKey = entry.getKey();
				long currenttime = System.currentTimeMillis();
	
				String[] tuplesp = entryKey.split("\\|",-1);
				String dip = tuplesp[0];
				
				if(upStreamoctets.get(entryKey)/(1024*1024*60) > 50) {//如果每秒流量大于50M则判断有攻击

					if(attackLasttimeMap.containsKey(entryKey)) {//如果存在就更新

						attackLasttimeMap.put(entryKey, currenttime);
						UpdateData(entry);
					}else {//不存在就新入数据
						attackStartMap.put(entryKey, currenttime);
						attackLasttimeMap.put(entryKey, currenttime);
						InsertIntoData(entry,currenttime);
					}
				}else {
					//如果没大于门限 也时间超过3分钟 则攻击结束
					if (attackLasttimeMap.containsKey(entryKey)) {
						if(currenttime - attackLasttimeMap.get(entryKey) > END_DURATION) {

							EndIntoData(entryKey);
							attackLasttimeMap.remove(entryKey);
							attackStartMap.remove(entryKey);
						}
					}
				}
			}

			// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束，避免漏处理数据
			ArrayList<String> removeItem = new ArrayList<String>();
			for(Entry<String,Long> entry:attackLasttimeMap.entrySet()) {
				String entryKey = entry.getKey();
				if(System.currentTimeMillis() - entry.getValue() > END_DURATION) {
					removeItem.add(entryKey);
					EndIntoData(entryKey);
					attackStartMap.remove(entryKey);
				}
			}

			for(String key  : removeItem) {
				attackLasttimeMap.remove(key);
			}

			removeItem.clear();
			upStreamoctets.clear();
			upStreampacket.clear();
			fixedSourceMap.clear();
		} catch (Exception e ) {
			e.printStackTrace();
			Logger.getLogger(getClass()).error("MetisBots Exception",e);;
		}
	}


	/**
	 * 新记录入库
	 * @Title: InsertIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	//	public void InsertIntoData() {
	public void InsertIntoData(Entry<String,Integer> entry,long time) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);

		String entryKey = entry.getKey();
		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String dport = tuplesp[2];
		String gis = tuplesp[3];
		String idc = tuplesp[4];

		String[] gisArray = gis.split("#");								
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];		
		String sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(time));
		String rowKeyAbnormal = accesstime + "_" + sip + "_"  + dip + "_" + dport + "_15";	
		 	

//		String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
//		String rowkey3 = String.format("%s_%s", accesstime, sip);

		String desc = String.format("僵尸主机,连接次数[%d],攻击峰值流量[%d KB/s]", fixedSourceMap.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));
		System.out.println("MetisBots----InsertIntoData" + rowKeyAbnormal  + " : " + desc);

		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEIP", sip);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTPORT", dport);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ABRNORMAL", "15");
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTIP", dip);							
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", fixedSourceMap.get(entryKey).toString());
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "EVALUATE", "40");
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));

//		hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
//		//			    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", "");
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "2");
//		hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);

		Date current = new Date();
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);

		collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
				sourceProvinceName, upStreampacket.get(entryKey), dateStr_hour,"BOTS", idc, sourceAreaCountry,(long)upStreamoctets.get(entryKey),(long)upStreampacket.get(entryKey)));


	}

	/**
	 * 更新记录字段
	 * @Title: UpdateData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param tuple
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void UpdateData(Entry<String,Integer> entry) {

		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);

		
		String entryKey = entry.getKey();
		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String dport = tuplesp[2];
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));
		String rowKeyAbnormal = accesstime + "_" + sip + "_"  + dip + "_" + dport + "_15";;
		String desc = String.format("僵尸主机 ,连接次数[%d],攻击峰值流量[%d KB/s]", fixedSourceMap.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));

		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);

		//		String rowkey3 = String.format("%s_%s", accesstime, sip);
		//		hbaseInstance.Add(ApolloProperties.SDS_ABNORMAL_LOG3, rowkey3, "cf", "ATTNUM", upStreampacket.get(entryKey).toString());
	}

	/**
	 * 攻击结束记录入库
	 * @Title: EndIntoData 
	 * @Description: TODO(这里用一句话描述这个方法的作用) 
	 * @param entryKey
	 * @return void    返回类型 
	 * @author 程彬
	 * @date 2016年11月21日 
	 */
	public void EndIntoData(String entryKey) {
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(
				zooserver);

		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String dport = tuplesp[2];
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackStartMap.get(entryKey)));

		//		String rowkey3 = String.format("%s_%s", accesstime, sip);
		String rowKeyAbnormal = accesstime + "_" + sip + "_"  + dip + "_" + dport + "_15";
//		String desc = String.format("僵尸主机结束 ,持续时间%s秒,连接次数[%d],攻击峰值流量[%d KB/s]", (attackLasttimeMap.get(entryKey)+60*1000l-attackStartMap.get(entryKey))/1000, fixedSourceMap.get(entryKey), upStreamoctets.get(entryKey)/(1024*60));
		String desc = String.format("僵尸主机结束 ,持续时间%s秒", (attackLasttimeMap.get(entryKey)+60*1000l-attackStartMap.get(entryKey))/1000);
		
		
//		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(upStreamoctets.get(entryKey)));	
//		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(upStreampacket.get(entryKey)));
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ENDTIME", new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(attackLasttimeMap.get(entryKey)+60*1000l)));

	}
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour",
				"abntype","idcno","sourcecountry","streamoctets","streampacket"));
		//		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport",
		//				"accesstime","sourcearea","sourcegeo","sourcecountry","destarea","destgeo",
		//				"evaluate","abnormal","desc","attnum","delaytime","flag"));

	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60); //10秒统计一次
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {

	}

}
